package data.platform.timescale.persistence.repository;

import data.platform.timescale.persistence.mapping.TagEO;
import io.r2dbc.spi.Batch;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.function.BiFunction;

@ConditionalOnBean(name = "timeScaleConfig")
@Slf4j
@Repository
@RequiredArgsConstructor
public class TsTagRepository {

    final DatabaseClient databaseClient;

    private final static String CREATE_SQL = "INSERT INTO tag(id, tag_json) values(%d, '%s')";

    private final static String QUERY_SQL = "SELECT * FROM tag WHERE tag_json = :tagJson";

    private final static String FIND_ALL_SQL = "SELECT * FROM tag";

    public static final BiFunction<Row, RowMetadata, TagEO> MAPPING_FUNCTION = (row, rowMetaData) -> TagEO.builder()
            .id(row.get("id", Integer.class))
            .tagJson(row.get("tag_json", String.class))
            .build();

    public Mono<Integer> saveBatch(Collection<TagEO> eos) {
        return databaseClient.inConnection(connection -> {
            Batch batch = connection.createBatch();
            for (TagEO eo : eos) {
                String sql = String.format(CREATE_SQL, eo.getId(), eo.getTagJson());
                batch = batch.add(sql);
            }
            return Flux.from(batch.execute())
                    .flatMap(result -> result.map((row, metadata) -> 1)) // 抛出异常到处理线程，否则无法抛出
                    .onErrorResume(error -> {
                        List<String> sqls = new ArrayList<>();
                        for (TagEO eo : eos) {
                            String sql = String.format(CREATE_SQL, eo.getId(), eo.getTagJson());
                            sqls.add(sql);
                        }
                        for (String sql : sqls) {
                            databaseClient.sql(sql).fetch().one().subscribe();
                        }
                        return Mono.empty();
                    })
                    .collectList().map(List::size);
        });
    }

    public Flux<TagEO> findAll() {
        return databaseClient.sql(FIND_ALL_SQL)
                .map(MAPPING_FUNCTION)
                .all();
    }

}
